package com.edu.hbade.test;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
import java.math.BigInteger;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;


/**
 * @author allen
 * @date 31/10/2017.
 */
public class HbaseDemo {

	static Configuration conf=null;
	static {
		conf= HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum","myhbase");
		conf.set("hbase.zookeeper.property.clientPort","2181");
		conf.set("log4j.logger.org.apache.hadoop.hbase","WARN");
	}

	public static void createTable(String tableName,String... families) throws Exception{
		HTableDescriptor tableDescriptor=new HTableDescriptor(TableName.valueOf(tableName));
		try{
			Connection connection=ConnectionFactory.createConnection(conf);
			Admin admin=connection.getAdmin();
			for(String family:families){
				tableDescriptor.addFamily(new HColumnDescriptor(family));
			}
			if(admin.tableExists(TableName.valueOf(tableName))){
				System.out.println("Table Exists");
				System.exit(0);
			}else{
				admin.createTable(tableDescriptor);
				System.out.println("Create table Success!!!Table Name:["+tableName+"]");
			}
		}catch (IOException e){
			e.printStackTrace();
		}
	}

	public static void deleteTable(String tableName){
		try{
			Connection connection=ConnectionFactory.createConnection(conf);
			Admin admin=connection.getAdmin();
			TableName table=TableName.valueOf(tableName);
			admin.disableTable(table);
			admin.deleteTable(table);
			System.out.println("delete table "+tableName+" ok!");
		}catch (IOException e){
			e.printStackTrace();
		}
	}


	public static void updateTable(String tableName,String rowKey,String familyName,String columnName,String value) throws Exception{
		try{
			Connection connection=ConnectionFactory.createConnection(conf);
			Table table=connection.getTable(TableName.valueOf(tableName));
			Put put=new Put(Bytes.toBytes(rowKey));
			put.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(columnName),Bytes.toBytes(value));
			table.put(put);
			System.out.println("Update table success");
		}catch (IOException e){
			e.printStackTrace();
		}
	}

	public static void addData(String rowKey,String tableName,String[] column,String[] value){
		try{
			Connection connection=ConnectionFactory.createConnection(conf);
			Table table=connection.getTable(TableName.valueOf(tableName));
			Put put=new Put(Bytes.toBytes(rowKey));
			HColumnDescriptor[] columnFamilies=table.getTableDescriptor().getColumnFamilies();
			for(int i=0;i<columnFamilies.length;i++){
				String familyName=columnFamilies[i].getNameAsString();
				if(familyName.equals("version")){
					for(int j=0;j<column.length;j++){
						put.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(column[j]),Bytes.toBytes(value[j]));
					}
					table.put(put);
					System.out.println("Add Data Success!");
				}
			}
		}catch(IOException e){

		}
	}



	public static void deleteAllColumn(String tableName,String rowKey){
		try{
			Connection connection=ConnectionFactory.createConnection(conf);
			Table table=connection.getTable(TableName.valueOf(tableName));
			Delete delAllColumn=new Delete(Bytes.toBytes(rowKey));
			table.delete(delAllColumn);
			System.out.println("Delete AllColumn Success");
		}catch(IOException e){
			e.printStackTrace();
		}
	}

	public static void deleteColumn(String tableName,String rowKey,String familyName,String columnName){
		try{
			Connection connection=ConnectionFactory.createConnection(conf);
			Table table=connection.getTable(TableName.valueOf(tableName));
			Delete delColumn=new Delete(Bytes.toBytes(rowKey));
			delColumn.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(columnName));
			table.delete(delColumn);
			System.out.println("Delete Column Success");
		}catch (IOException e){
			e.printStackTrace();
		}
	}

	public static void scanResult(String tableName){
		byte[] POSTFIX=new byte[]{0x00};
		Scan scan=new Scan();
//		scan.setBatch(2);
//		Filter filter=new RowFilter(CompareFilter.CompareOp.EQUAL,new SubstringComparator(".*+1502176018+20900"));
//		ArrayList<Filter> filterArrayList=new ArrayList<Filter>();
//		Filter filterRegex=new RowFilter(CompareFilter.CompareOp.EQUAL,new RegexStringComparator("^.*\\+1502176018\\+20900$"));
//		Filter filter=new RowFilter(CompareFilter.CompareOp.EQUAL,new RegexStringComparator("^.*\\+1502176018\\+.*$"));

//
//		Filter filterPage=new PageFilter(2);
////		Filter filterPage=new ColumnPaginationFilter(5,0);
//		scan.setFilter(filterPage);
//		byte[] startRow=Bytes.add("24b1f27c982ea87XyZf+1502176018+209001".getBytes(),POSTFIX);
//		scan.setStartRow(startRow);

//		scan.setRowOffsetPerColumnFamily(4);
//		scan.setMaxResultsPerColumnFamily(2);

//		scan.setStartRow("24b1f27c982ea87XyZf+1502176018+180".getBytes());
//		scan.setMaxResultsPerColumnFamily(2);
//		filterArrayList.add(filterRegex);
//		filterArrayList.add(filterPage);
//		Filter filterList=new FilterList(FilterList.Operator.MUST_PASS_ALL,filterArrayList);
//		scan.setFilter(filterList);
//        scan.

//		scan.setMaxResultSize(1);

//		scan.setReversed(true);
//		scan.setStartRow("#.1502176018+20900".getBytes());
//		scan.setStopRow(":.1502176018+20900".getBytes());
		try{
			Connection connection=ConnectionFactory.createConnection(conf);
			Table table=connection.getTable(TableName.valueOf(tableName));
			ResultScanner rs=table.getScanner(scan);
			for(Result r:rs){
				for(Cell cell:r.listCells()){
					System.out.println("rowKey:"+Bytes.toString(cell.getRowArray(),cell.getRowOffset(),cell.getRowLength()));
					System.out.println("family:"+Bytes.toString(cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength()));
					System.out.println("qualifier:"+Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()));
					System.out.println("value:"+Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()));
					System.out.println("Timestamp:"+cell.getTimestamp());

				}

				System.out.println("----------------------");
			}
		}catch (IOException e){
			e.printStackTrace();
		}
	}

	public static void scanGetPage(String tableName,int offset,int limit){
		try{
			Scan scan =new Scan();
			Connection connection=ConnectionFactory.createConnection(conf);
			Table table=connection.getTable(TableName.valueOf(tableName));
			ResultScanner rs=table.getScanner(scan);
			int count=0;
			for(Result r:rs){
				if(++count<=offset){
					continue;
				}
				for(Cell cell:r.listCells()){
					System.out.println("rowKey:"+Bytes.toString(cell.getRowArray(),cell.getRowOffset(),cell.getRowLength()));
					System.out.println("family:"+Bytes.toString(cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength()));
					System.out.println("qualifier:"+Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()));
					System.out.println("value:"+Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()));
					System.out.println("Timestamp:"+cell.getTimestamp());

				}

				if(count==offset+limit) {
					break;
				}

				System.out.println("----------------------");
			}
		}catch (IOException e){
			e.printStackTrace();
		}
	}

	public  static void getResultByVersion(String tableName,String rowKey,String familyName,String columnName){
		try{
			Connection connection=ConnectionFactory.createConnection(conf);
			Table table =connection.getTable(TableName.valueOf(tableName));
			Get get =new Get(Bytes.toBytes(rowKey));
			get.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(columnName));
			get.setMaxVersions(3);
			Result result=table.get(get);
			for(Cell cell :result.listCells()){
				System.out.println("family:"+Bytes.toString(cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength()));
				System.out.println("qualifier:"+Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()));
				System.out.println("value:"+Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()));
				System.out.println("Timestamp:"+cell.getTimestamp());
				System.out.println("---------------");
			}
		}catch (IOException e){
			e.printStackTrace();
		}
	}


	public static void scanTimeRange(String tableName,long  beginTime,long endTime) throws ParseException {
//		SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//		long a=formatter.parse(time).getTime();
//		System.out.println("time parse is "+a);

		try{
			Scan scan=new Scan();
//			scan.setTimeRange(beginTime,endTime);
//			Filter filter= new ValueFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL,new BinaryComparator("2017-11-08 14".getBytes()) );
			Filter filter= new ValueFilter(CompareFilter.CompareOp.LESS_OR_EQUAL,new BinaryComparator("2017-11-08 15".getBytes()) );
			scan.setFilter(filter);
			scan.setReversed(true);
			Connection connection=ConnectionFactory.createConnection(conf);
			Table table=connection.getTable(TableName.valueOf(tableName));
			ResultScanner rs=table.getScanner(scan);


			for(Result r:rs) {

				for (Cell cell : r.listCells()) {
					System.out.println("rowKey:" + Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
					System.out.println("family:" + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
					System.out.println("qualifier:" + Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
					System.out.println("value:" + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
					System.out.println("Timestamp:" + cell.getTimestamp());

				}

				System.out.println("----------------------");
			}
		}catch (IOException e){
			e.printStackTrace();
		}
	}

	public static String nowTime(){
		SimpleDateFormat df=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		return df.format(new Date());
	}

	public static  long timeRetain(){
		return Long.MAX_VALUE-System.currentTimeMillis();
	}

	public static void main(String[] args) throws Exception {
		System.setProperty("hadoop.home.dir", "/usr/local/hadoop");
//		String[] families={"create_time","update_time","downtime"};
//		HbaseDemo.createTable("ota_record",families);
//		HbaseDemo.createTable("ota_pre_record","version");
//		HbaseDemo.deleteTable("ota_pre_record");


		nowTime();
//		HbaseDemo.updateTable("ota_pre_record","324b1f27c982ea87XyZf+1502176018+20900+"+retain,
//				"version","check_time",nowTime());
//
//		HbaseDemo.updateTable("ota_pre_record",timeRetain()+"+324b1f27c982ea87XyZf+1502176018+20900",
//				"version","check_time",nowTime());
//		HbaseDemo.updateTable("ota_pre_record","324b1f27c982ea87XyZf+1502176018+20900","version","download_time","2017-11-20 15:37:34");
//		HbaseDemo.updateTable("ota_pre_record","324b1f27c982ea87XyZf+1502176018+20900+"+timeRetain(),"version","check_time",nowTime());
//		HbaseDemo.updateTable("ota_pre_record","324b1f27c982ea87XyZf+1502176018+20900","version","download_time","2017-11-16 14:37:34");
//		HbaseDemo.updateTable("ota_pre_record","24b1f27c982ea87XyZf+1502176018+18900","version","upgrade_time","2017-11-18 14:35:34");
//		HbaseDemo.deleteColumn("ota_pre_record","24b1f27c982ea87XyZf+1502176018+20900","version","upgrade_time");
//		HbaseDemo.deleteAllColumn("ota_pre_record","24b1f27c982ea87XyZf+1502176018+20900");
//		HbaseDemo.getResultByVersion("ota_pre_record","24b1f27c982ea87XyZf+1502176018+20900","version","check_time");

		HbaseDemo.scanResult("ota_pre_record");
//		HbaseDemo.scanGetPage("ota_pre_record",1,2);


//		Long beginTime=1510126004791;
//		scanTimeRange("ota_pre_record",1510123118901L,1510126004791L);
	}
}
